package com.zhang.flink.example;

import com.zhang.flink.bean.AdCount;
import com.zhang.flink.bean.AdsClickLog;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @title: 每隔五分钟统计过去一个小时各省份广告点击量，
 * 同时对同一个用户点击同一个广告超过100次拉入黑名单处理，后续不再进行总数统计
 * @author: zhang
 * @date: 2021/12/12 21:48
 */
public class AdClickAnalysis {
    public static void main(String[] args) throws Exception {
        // todo 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // todo 获取数据转化 pojo class 并分配时间戳和水位线
        SingleOutputStreamOperator<AdsClickLog> adsClickLogDS = env
                .readTextFile("src/main/resources/AdClickLog.csv")
                .map(new MapFunction<String, AdsClickLog>() {
                    @Override
                    public AdsClickLog map(String value) throws Exception {
                        String[] fields = value.split(",");
                        return new AdsClickLog(
                                fields[0], fields[1], fields[2], fields[3],
                                Long.parseLong(fields[4])
                        );
                    }
                })
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<AdsClickLog>forMonotonousTimestamps()
                                .withTimestampAssigner(new SerializableTimestampAssigner<AdsClickLog>() {
                                    @Override
                                    public long extractTimestamp(AdsClickLog element, long recordTimestamp) {
                                        return element.ts * 1000L;
                                    }
                                })
                );
        // todo 把有刷单行为的用户信息输出到黑名单(侧输出流中)并做过滤
        SingleOutputStreamOperator<AdsClickLog> warningDS = adsClickLogDS
                .keyBy(r -> r.userId + "_" + r.adId)
                .process(new FilterUserBlackListResult(100L));
        // todo 每隔五分钟统计过去一个小时各省份广告点击量
        SingleOutputStreamOperator<AdCount> result = warningDS
                .keyBy(r -> r.province)
                .window(SlidingEventTimeWindows.of(Time.hours(1), Time.minutes(5)))
                .aggregate(new AdCountAgg(), new AdWindowResult());

        // todo 提取侧输出流报警数据
        warningDS.getSideOutput(new OutputTag<String>("warning") {
        }).print("warning:");
        // todo 打印各省份点击情况
        result.print();

        // todo 启动任务
        env.execute();

    }

    public static class AdWindowResult extends ProcessWindowFunction<Long, AdCount, String, TimeWindow> {
        @Override
        public void process(String key, ProcessWindowFunction<Long, AdCount, String, TimeWindow>.Context context, Iterable<Long> elements, Collector<AdCount> out) throws Exception {
            out.collect(new AdCount(
                    key,
                    elements.iterator().next(),
                    context.window().getStart(),
                    context.window().getEnd()
            ));
        }
    }

    public static class AdCountAgg implements AggregateFunction<AdsClickLog, Long, Long> {
        @Override
        public Long createAccumulator() {
            return 0L;
        }

        @Override
        public Long add(AdsClickLog value, Long accumulator) {
            return accumulator + 1L;
        }

        @Override
        public Long getResult(Long accumulator) {
            return accumulator;
        }

        @Override
        public Long merge(Long a, Long b) {
            return a + b;
        }
    }

    public static class FilterUserBlackListResult extends KeyedProcessFunction<String, AdsClickLog, AdsClickLog> {
        private Long maxClickCount;

        public FilterUserBlackListResult(Long maxClickCount) {
            this.maxClickCount = maxClickCount;
        }

        // 保存点击当前用户对某一个广告点击次数
        private ValueState<Long> countState;
        //判断用户是否在黑名单
        private ValueState<Boolean> isBlackList;


        @Override
        public void open(Configuration parameters) throws Exception {
            countState = getRuntimeContext().getState(
                    new ValueStateDescriptor<Long>(
                            "count-state", Types.LONG
                    )
            );
            isBlackList = getRuntimeContext().getState(
                    new ValueStateDescriptor<Boolean>(
                            "isBlackList", Types.BOOLEAN
                    )
            );

        }

        @Override
        public void processElement(AdsClickLog value, KeyedProcessFunction<String, AdsClickLog, AdsClickLog>.Context ctx, Collector<AdsClickLog> out) throws Exception {
            //取出点击状态变量
            Long count = countState.value();
            if (count == null) {
                // 注册定时器，0点清除
                countState.update(1L);
                long ts = (ctx.timerService().currentProcessingTime() / (1000 * 60 * 60 * 24) + 1) * (24 * 60 * 60 * 1000L) - 8 * 60 * 60 * 1000L;
                ctx.timerService().registerProcessingTimeTimer(ts);
            } else {

                count = count + 1L;
                countState.update(count);
                //当点击次数超过阈值时，
                if (count >= maxClickCount) {
                    //判断是否已经加入黑名单报警，避免重复报警
                    if (isBlackList.value() == null) {
                        ctx.output(new OutputTag<String>("warning") {
                        }, value.userId + "点击了" + value.adId + "广告达到" + maxClickCount + "次，存在恶意点击来入黑名单，报警！！！");
                        //加入黑名单
                        isBlackList.update(true);
                    }
                    //如果点击次数大于报警阈值则直接过滤掉，不进行下面的统计
                    return;
                }
            }
            //如果没有返回值，正常输出数据到主流
            out.collect(value);
        }

        @Override
        public void onTimer(long timestamp, KeyedProcessFunction<String, AdsClickLog, AdsClickLog>.OnTimerContext ctx, Collector<AdsClickLog> out) throws Exception {
            countState.clear();
            isBlackList.clear();
        }
    }
}
